package com.hu.flink12.api.source;

import com.hu.common.Constants;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.Tuple2;

/**
 * @Author: hujianjun
 * @Date: 2021/2/3 22:45
 * @Describe: 自定义每5s查询数据
 */
public class MysqlJDBCSink {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //TODO 2.读取数据源
        DataStream<String> inputStream = env.fromElements("1 hjj@iyunbao.com", "2 hjj2@iyunbao.com");

        //TODO 3.转换操作
        DataStream<Tuple2<Integer, String>> dataStream = inputStream.map(new MapFunction<String, Tuple2<Integer, String>>() {
            @Override
            public Tuple2<Integer, String> map(String s) throws Exception {
                String[] arr = s.split(" ");
                return new Tuple2<>(Integer.parseInt(arr[0]), arr[1]);
            }
        });

        //TODO 4.sink操作
        dataStream.addSink(JdbcSink.sink("insert into test.user2 values(?,?)"
                , (ps, v) -> {
                    ps.setInt(1, v._1);
                    ps.setString(2, v._2);
                }, new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:3306/test?useUnicode=true&serverTimezone=UTC&useSSL=false&allowPublicKeyRetrieval=true")
                        .withDriverName(Constants.MYSQL_DRIVER_NAME)
                        .withUsername("root")
                        .withPassword("root")
                        .build()));

        //TODO 5.执行程序
        env.execute();
    }
}
